package org.example.hotcommodity;

import org.example.hotcommodity.ItemViewCount;
import org.example.hotcommodity.UserBehavior;
import org.apache.commons.compress.utils.Lists;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.timestamps.AscendingTimestampExtractor;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.example.ordermonitor.OrderEvent;

import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Comparator;
import java.util.Properties;

public class HotItems {
    public static void main(String[] args) throws Exception {
        //创建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        DataStream<UserBehavior> dataSource = env.fromCollection(Arrays.asList(
                new UserBehavior(1L, 1L, 1, "buy", 1558430841L),
                new UserBehavior(2L, 1L, 2, "pv", 1558430842L),
                new UserBehavior(2L, 2L, 3, "buy", 1558430843L)
        ));
        DataStream<UserBehavior> dataStream=dataSource
                .assignTimestampsAndWatermarks(
                        new AscendingTimestampExtractor<UserBehavior>() { //升序
                            @Override
                            public long extractAscendingTimestamp(UserBehavior userBehavior) {//获取事件时间戳，秒级转毫秒级
                                return userBehavior.getTimestamp()*1000L;
                            }
                        });

        //分组聚合得到结果数据
        DataStream<ItemViewCount> aggStream=dataStream
                .filter(data -> "pv".equals(data.getBehavior())) //过滤“pv”行为
                .keyBy(UserBehavior::getItemId)
                .timeWindow(Time.minutes(60),Time.minutes(5)) //每5分钟更新一次1小时窗口数据
                .aggregate(new ItemCountAgg(),new WindowItemCountResult());

        //收集同一窗口所有商品的count数据，按top 5输出
        DataStream resultDs=aggStream
                .keyBy("windowEnd")
                .process(new TopNItems(5));



        //输出并执行
        resultDs.print("每隔五分钟最近一小时前五的热门商品");
        env.execute("hot items analysis");
    }


    //泛型1:输入类型   泛型2:聚合状态类型   泛型3:输出类型
    public static class ItemCountAgg implements AggregateFunction<UserBehavior,Long,Long>{

        @Override
        public Long createAccumulator() {//创建累加器并给初始值
            return 0L;
        }

        @Override
        public Long add(UserBehavior userBehavior,Long accumulator) {//每次计算累加器加一,并返回新的累加器值
            return accumulator+1;
        }

        @Override
        public Long getResult(Long accumulator) {//累加器最终给外部返回的值
            return accumulator;
        }

        @Override
        public Long merge(Long a, Long b) { //合并两个累加器，返回合并后的累加器的状态，这儿用不到.用不到.
            return a+b;
        }
    }

    //参数1：输入类型，即ItemCountAgg的输出类型  参数2：输出类型  参数3：keyBy的返回值键值对中value的类型  参数4： 窗口类型
    public static class WindowItemCountResult implements WindowFunction<Long,ItemViewCount, Long, TimeWindow>{

        //迭代器中装的是输入类型
        @Override
        public void apply(Long key, TimeWindow window, Iterable<Long> iterable, Collector<ItemViewCount> collector) throws Exception {
            //包装成一个ItemViewCount对象输出
            collector.collect(new ItemViewCount(key.longValue(),window.getEnd(),iterable.iterator().next()));
        }
    }

    //参数1：keyBy返回值类型  参数2：输入类型  参数3：输出类型
    public static class TopNItems extends KeyedProcessFunction<Tuple,ItemViewCount,String>{
        private Integer topSize;
        private ListState<ItemViewCount> listState; //列表状态，保存当前窗口所有输出的ItemViewCount

        public TopNItems(Integer topSize) {
            this.topSize = topSize;
        }

        @Override
        public void open(Configuration parameters) throws Exception {
            listState =getRuntimeContext().getListState(new ListStateDescriptor<ItemViewCount>("item-view-count-list",ItemViewCount.class));
        }

        //每来一条数据如何处理
        @Override
        public void processElement(ItemViewCount value, Context context, Collector<String> collector) throws Exception {
            //每来一条数据，存入List中，并注册定时器（只有触发时间一样，定时器就是同一个）
            listState.add(value);
            context.timerService().registerEventTimeTimer(value.getWindowEnd());//注册定时器
        }

        //定时器触发时的逻辑
        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
            //转换成Arraylist再排序
            ArrayList<ItemViewCount> itemViewCounts = Lists.newArrayList(listState.get().iterator());

            itemViewCounts.sort(new Comparator<ItemViewCount>() {
                @Override
                public int compare(ItemViewCount o1, ItemViewCount o2) {//前大于后返回负数，为倒序
                    if(o1.getCount() > o2.getCount())
                        return -1;
                    else if (o1.getCount() == o2.getCount())
                        return 0;
                    else
                        return 1;
                }
            });

            //定义输出结果格式
            StringBuilder resultBuilder=new StringBuilder();
            resultBuilder.append("===================\n");
            resultBuilder.append("窗口结束时间：").append(new Timestamp(timestamp)).append("\n"); //输出windowend

            //遍历输出
            for (int i = 0; i < Math.min(topSize,itemViewCounts.size()); i++) {
                ItemViewCount currentItemViewCount = itemViewCounts.get(i);
                resultBuilder.append("Number").append(i+1).append(":")
                        .append("商品ID：").append(currentItemViewCount.getItemId())
                        .append("浏览量：").append(currentItemViewCount.getCount())
                        .append("\n");
            }

            resultBuilder.append("===================\n\n");

            Thread.sleep(1000L);//控制输出频率
            out.collect(resultBuilder.toString());
            listState.clear();//清空状态
        }

        @Override
        public void close() throws Exception {
            listState.clear();//清空状态
        }
    }
}

